%%%-------------------------------------------------------------------
%%% File    : mapred.erl
%%% Author  :  <vjache>
%%% Description : This module implements a Map/Reduce computation 
%%%				  paradigm introduced by Google but more powerfull and 
%%%				  flexibile. The programming style with this framework
%%%				  suggests the intensive use of iter & pipe Lego's.
%%%				  
%%%				  This implementation has the following advantages:
%%%					1. Map/Reduce phases may form hierarcle cascades
%%%					2. Reduce operation is join friendly i.e. allows 
%%%					   write routines for INNER JOIN and OUTER JOIN 
%%%					   much more easily (in conjunction with 'iter' Lego).
%%%
%%% Created :  8 Jul 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(mapred).

-include_lib("eunit/include/eunit.hrl").

-include("../include/rrs.hrl").
-include("../include/pgt.hrl").
-include("../include/iter.hrl").
-include("../include/pipe.hrl").

%% API
%% User functions
-export([exec/1,
		 exec/2,
		 aexec/1,
		 aexec/2,
		 iter/1,
		 execf/1,
		 execf/2,
		 aexecf/1,
		 aexecf/2,
		 cancel/1,
		 cancel/2,
		 key2node/1]).

%% System funcions not intended for user
-export([red_proc_init/1,
		 map_proc_init/2]).

%% Special non eunit test functions
-export([test_bulk_join/0,
		 test_bulk_join2/0,
		 test_bulk_join_zip/0]).

-define(IS_RED_SPEC(RedSpec),element(1,RedSpec)==red).
-define(IS_MAP_SPEC(MapSpec),element(1,MapSpec)==map).
-define(IS_SPEC(Spec),(element(1,Spec)==red orelse element(1,Spec)==map)).

-record(job_handle,{caller_pid=self(),job_pid,job_ref}).
-record(red_conf,{id,map_ids,red_fun,opts=[]}).
-record(map_conf,{id,map_fun,opts=[]}).
%%====================================================================
%% API
%%====================================================================

%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
exec(Spec) when ?IS_SPEC(Spec) ->
	wait_finish(aexec(Spec)).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
exec(Spec, OutPipe) 
  when ?IS_SPEC(Spec),
	   is_record(OutPipe, pipe) ->
	wait_finish(aexec(Spec,OutPipe)).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
aexec(MapSpec) 
  when ?IS_MAP_SPEC(MapSpec) ->
	aexec({red,nil,[MapSpec]});
aexec(RedSpec) when ?IS_RED_SPEC(RedSpec) ->
	Pgt=red_make(RedSpec,[],make_ref()),
	{JobPid,JobRef}=pgt:new_monitor(Pgt),
	#job_handle{caller_pid=self(), job_pid=JobPid, job_ref=JobRef}.
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
aexec(MapSpec, OutPipe) 
  when ?IS_MAP_SPEC(MapSpec),
	   is_record(OutPipe, pipe) ->
	aexec({red,nil,[MapSpec]}, OutPipe);
aexec(RedSpec, OutPipe) 
  when ?IS_RED_SPEC(RedSpec),
	   is_record(OutPipe, pipe) ->
	%% 	{output,OutPipe}
	RedSpec1=case RedSpec of
				 {red,RedFun,MapSpecs,Opts} ->
					 {red,RedFun,MapSpecs,[{output,OutPipe}|Opts]};
				 {red,RedFun,MapSpecs} ->
					 {red,RedFun,MapSpecs,[{output,OutPipe}]}
			 end,
	Pgt=red_make(RedSpec1,[],make_ref()),
	{JobPid,JobRef}=pgt:new_monitor(Pgt),
	#job_handle{caller_pid=self(), job_pid=JobPid, job_ref=JobRef}.
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
iter(Spec) when ?IS_SPEC(Spec) ->
	#iter{param=Spec,
		  impl=#fun_set{
						open_fun=fun iter_open/1,
						next_fun=fun iter_next/1,
						close_fun=fun iter_close/1}
		 }.
iter_open(Spec)-> 
	Self=self(),
	StPid=spawn_link(
			fun()->
					QPid=bqueue:create(100),
					JobHandle=aexec(Spec, bqueue:new_pipe(QPid)),
					Self ! {self(),{JobHandle,QPid}},
					try wait_finish(JobHandle) of
						{mapred_cancel,_CancelReason} ->
							bqueue:destroy(QPid, true);
						_ ->
							bqueue:destroy(QPid, false)
					catch
						_:Err ->
							bqueue:destroy(QPid, true),
							io:format("ERRRRRR ~p~n",[Err]),
							throw(Err)
					end
			end),
	receive
		{StPid,{JobHandle,QPid}} ->
			{JobHandle, iter:open(bqueue:new_iter(QPid))}
	end.
iter_next({JobHandle,IterState})->
	{IterState1,Item}=iter:next(IterState),
	case Item of
		?EOF ->
			iter:close(IterState1),
			{{JobHandle,?EOF},Item};
		_ ->
			{{JobHandle,IterState1},Item}
	end.
iter_close({_JobHandle,?EOF})->
	ok;
iter_close({JobHandle,IterState})->
	cancel(JobHandle,iter_close),
	iter:close(IterState).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
execf(RedSpecFile) when is_list(RedSpecFile) ->
	throw(not_implemented_yet).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
execf(RedSpecFile, OutPipe) 
  when is_list(RedSpecFile),
	   is_record(OutPipe, pipe) ->
	throw(not_implemented_yet).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
aexecf(RedSpecFile) when is_list(RedSpecFile) ->
	throw(not_implemented_yet).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
aexecf(RedSpecFile, OutPipe) 
  when is_list(RedSpecFile),
	   is_record(OutPipe, pipe) ->
	throw(not_implemented_yet).

cancel(JobHandle) ->
	cancel(JobHandle,kill).	
cancel(JobHandle,Reason) 
  when is_record(JobHandle, job_handle) ->
	true=exit(JobHandle#job_handle.job_pid, {mapred_cancel,Reason}),
	ok.
wait_finish(JobHandle) ->
	wait_finish(JobHandle, infinity).
wait_finish(JobHandle, Timeout)
  when is_record(JobHandle, job_handle) and
	   (Timeout==infinity orelse 
			(is_integer(Timeout) and Timeout>=0)) ->
	MonitorRef=JobHandle#job_handle.job_ref,
	Object=JobHandle#job_handle.job_pid,
	CallerPid=JobHandle#job_handle.caller_pid,
	receive
		{'DOWN', MonitorRef, process, Object, noproc} 
		  when CallerPid==self()  ->
			throw({no_such_job,JobHandle});
		{'DOWN', MonitorRef, process, Object, noproc} ->
			no_such_job;
		{'DOWN', MonitorRef, process, Object, noconnection} ->
			throw({noconnection,JobHandle});
		{'DOWN', MonitorRef, process, Object, normal} ->
			ok;
		{'DOWN', MonitorRef, process, Object, Reason} ->
			case pgt:unwrap_error(Reason) of
				{mapred_cancel,_CancelReason}=CancelMsg ->
					CancelMsg;
				_ ->
					throw({mapred_failed, Reason})
			end
	after
			Timeout ->
			timeout
	end.


%%====================================================================
%% Internal functions
%%====================================================================
red_make({red, RedFun, MapSpecList},ConfList,Id) ->
	red_make({red, RedFun, MapSpecList,[]},ConfList,Id);
red_make({red, RedFun, MapSpecList,Opts},ConfList,Id) 
  when (is_function(RedFun,2) orelse RedFun==nil),
	   is_list(MapSpecList),
	   is_list(ConfList),
	   is_list(Opts) ->
	IdMapSpecList=[{make_ref(),MapSpec}||MapSpec<-MapSpecList],
	MapIdList=[MapId||{MapId,_MapSpec}<-IdMapSpecList],
	RedConf=#red_conf{
			  id=Id,
			  map_ids=MapIdList,
			  red_fun=RedFun,
			  opts=Opts
			  },
	#pg{id=Id,
		name="Reduce group",
		pgs=[map_make(MapSpec,[RedConf| ConfList],MapId)||{MapId,MapSpec}<-IdMapSpecList]}.
map_make({map, MapFun, Iter},ConfList,Id) ->
	map_make({map, MapFun, Iter,[]},ConfList,Id);
map_make({map, MapFun, Iter,Opts},ConfList,Id)
  when is_record(Iter, iter),
	   is_function(MapFun, 1),
	   is_list(ConfList) ->
	map_make({map, MapFun, {node(),Iter},Opts},ConfList,Id);
map_make({map, MapFun, {Node,Iter},Opts},ConfList,Id) 
  when is_record(Iter, iter),
	   is_function(MapFun, 1),
	   is_list(ConfList) ->
	MapConf=#map_conf{
			  id=Id,
			  map_fun=MapFun,
			  opts=Opts},
	#pg{id=Id,
		name="Leaf map group",
		workers=[#worker{
						 spawn_spec={
									 Node, 
									 ?MODULE, 
									 map_proc_init, 
									 [Iter, [MapConf |ConfList]]
									}
						}
				]
	   };
map_make({map, MapFun, RedSpec, Opts},ConfList,Id) 
  when is_function(MapFun, 1),
	   element(1,RedSpec)== red,
	   is_list(ConfList) ->
	MapConf=#map_conf{
			  id=Id,
			  map_fun=MapFun,
			  opts=Opts},
	#pg{id=Id,
		name="Map group",
		pgs=[red_make(RedSpec,[MapConf| ConfList],make_ref())]
	   }.
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Map process funs
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% map_proc_init(Iter, ConfList) ->
%% 	map_proc_loop(iter:open(Iter),ConfList).
%% 
%% map_proc_loop(IterState,
%% 			  [#map_conf{id=Id,
%% 						 map_fun=MapFun,
%% 						 opts=_Opts}|ConfListTail]=ConfList)
%%   when is_function(MapFun, 1) ->
%% 	case iter:next(IterState) of
%% 		{IterState1,?EOF} ->
%% 			iter:close(IterState1);
%% 		{IterState1,Item} -> 
%% 			ForeachFun=fun(MappedItem={Key,_Item1})->
%% 							   Node=key2node(Key),
%% 							   RedPid=spawn_reducer(Node,ConfListTail),
%% 							   RedPid ! {item,Id,MappedItem}
%% 					   end,
%% 			case MapFun(Item) of
%% 				MappedItemList when is_list(MappedItemList) ->
%% 					lists:foreach(ForeachFun, MappedItemList);
%% 				MappedItemIter when is_record(MappedItemIter, iter) ->
%% 					{IterFinalState,ok}=iter:foreach(MappedItemIter,ForeachFun),
%% 					iter:close(IterFinalState);
%% 				MappedItem ->
%% 					ForeachFun(MappedItem)
%% 			end,
%% 			map_proc_loop(IterState1,ConfList)
%% 	end.

map_proc_init(Iter, 
			  [#map_conf{id=_Id,
						 map_fun=MapFun,
						 opts=_Opts}|_ConfListTail]=ConfList) ->
	MappedItemsIter=
		iter_concat:new(
		  iter_mapper:new(
			Iter, 
			fun(Item)->
					case MapFun(Item) of
						MappedItemList when is_list(MappedItemList) ->
							iter:from_list(MappedItemList);
						MappedItemIter when is_record(MappedItemIter, iter) ->
							MappedItemIter;
						MappedItem ->
							MappedItem
					end
			end)),
	Pipe=map_send_pipe(ConfList),
	Pump=pump:new(MappedItemsIter, Pipe),
	pump:init(Pump).

map_send_pipe([#map_conf{id=Id,
						 map_fun=_MapFun,
						 opts=Opts}|ConfListTail]=_ConfList) ->
	
	pipe_switch:new(
	  fun(_MappedItem={Key,_Item1})->
			  _Node=key2node(Key)
	  end, 
	  fun(Node)->
			  SenderPipe=pipe:simple_immutable(
						   fun()->
								   _RedPid=spawn_reducer(Node,ConfListTail)
						   end, 
						   fun(RedPid,DataItem)->
								   RedPid ! {item,DataItem}
						   end, 
						   fun(RedPid)->
								   RedPid ! {mapper_finish,Id,self()},
								   receive
									   {acknowledged,RedPid} ->
										   ok
								   end
						   end),
			  ZipPipe=case proplists:get_value(zip, Opts) of
						  0 ->
							  SenderPipe;
						  undefined ->
							  SenderPipe;
						  BuffSize when is_integer(BuffSize) andalso BuffSize>0 ->
							  pipe_pzip:new(BuffSize, SenderPipe);
						  BadVal ->
							  throw({bad_option_value,{zip,BadVal}})
					  end,
			  _TaggerPipe=pipe_mapper:new(
							fun(MappedItem)->
									{Id,MappedItem}
							end, ZipPipe)
	  end).
	
spawn_reducer(Node,[RedConf| _ConfListTail]=ConfList)
  when is_record(RedConf, red_conf)->
	RedId=RedConf#red_conf.id,
	case get(RedGlobalId={'$red-wrk',Node,RedId}) of
		undefined ->
			RedPid=pgt:ppg_obtain_worker(
					 RedGlobalId, Node, ?MODULE, red_proc_init, [ConfList]),
			put(RedGlobalId,RedPid),
			RedPid;
		RedPid ->
			RedPid
	end.

key2node(Key) ->
	KRSKey='$key_route_snapshot',
	case get(KRSKey) of
		undefined ->
			Snap=svc_rrs:make_route_snapshot(),
			put(KRSKey,Snap),
			?ROUTE2UPNODE(svc_rrs:key_route(Key,Snap));
		Snap ->
			?ROUTE2UPNODE(svc_rrs:key_route(Key,Snap))
	end.
%% 	node().


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Reduce process funs
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
red_proc_init([RedConf|ConfListTail]=_ConfList)
  when is_record(RedConf, red_conf) andalso 
		   RedConf#red_conf.red_fun==nil ->
	%% Receive & collect data from mappers
	red_proc_rcv_pump(reducer2mapper_pipe(RedConf,ConfListTail));
red_proc_init([RedConf|ConfListTail]=_ConfList)
  when is_record(RedConf, red_conf) ->
	RedOpts=RedConf#red_conf.opts,
	%% Receive & collect data from mappers
	red_proc_rcv_pump(new_accum_pipe(RedOpts)),
	%% Compute a set of all keys
	KeyIter=new_accum_key_iter(),
	%% Create pipe to mapper 
	R2MPipe=reducer2mapper_pipe(RedConf,ConfListTail),
	%% Pump accumulated keys to pipe
	R2MPump=pump:new(KeyIter, R2MPipe),
	pump:init(R2MPump).

%% direct_mapper_pipe([MapConf|_]=ConfListTail) when 
%%   is_record(MapConf,map_conf) ->
%% 	Node=node(),
%% 	MapperId=MapConf#map_conf.id,
%% 	bqueue:new_pipe_one2many_pattern(
%% 	  100, 
%% 	  fun(QueueIter) ->
%% 			  Iter=iter_mapper:new(
%% 					 QueueIter, 
%% 					 fun({_Id,Item})->
%% 							 Item
%% 					 end),
%% 			  pgt:ppg_obtain_worker(
%% 				_WorkId={Node,MapperId}, 
%% 				Node, ?MODULE, map_proc_init, 
%% 				[Iter,ConfListTail]) 
%% 	  end).
reducer2mapper_pipe(RedConf,ConfList) ->
	RedOpts=RedConf#red_conf.opts,
	case proplists:get_value(concurrent, RedOpts) of
		undefined ->
			reducer2mapper_pipe_seq(RedConf,ConfList);
		Atom when Atom==default orelse
				  Atom==schedulers ->
			Num=erlang:system_info(schedulers),
			reducer2mapper_pipe_con(Num,RedConf,ConfList);
		Num when is_integer(Num),
				 Num>1 ->
			reducer2mapper_pipe_con(Num,RedConf,ConfList);
		BadVal ->
			throw({bad_option_value,{concurrent,BadVal}})
	end.
%% Sequential RedFun(Key,[Iter]) apply (concurency 1) 
reducer2mapper_pipe_seq(RedConf,ConfListTail) ->
	reducer2mapper_pipe_con(1,RedConf,ConfListTail).
%% Cincurrent RedFun(Key,[Iter]) apply (concurency >1)
reducer2mapper_pipe_con(Concurency,RedConf,ConfListTail) ->
	RedOpts=RedConf#red_conf.opts,
	RedFun=RedConf#red_conf.red_fun,
	MapIdList=RedConf#red_conf.map_ids,
	QPipe=bqueue:new_pipe_one2many_pattern(
			1000, 
			fun(QueueIter) ->
					%% Create iter that applies reduces
					TotalIter=case RedFun of
								  nil ->
									  iter_mapper:new(
										QueueIter, 
										fun({_Id,{_Key,Item}})->
												Item
										end);
								  _ ->	  
									  iter_concat:new(
										iter_mapper:new(
										  QueueIter, 
										  fun({Key,IterList}=_RedArgs)->
												  Result=RedFun(Key,IterList),
												  iter_cache:clear(),
												  Result
										  end))
							  end,
					Node=node(),
					PidList=case ConfListTail of %% Decide what to do further: spawn next map | finish
						[MapConf|_] 
						  when is_record(MapConf,map_conf) ->
							%% Create iter that applies reduces & concat results to single iterator
							MapperId=MapConf#map_conf.id,
							[ pgt:ppg_obtain_worker(
										_WorkId={Node,MapperId,N}, 
										Node, ?MODULE, map_proc_init, 
										[TotalIter,ConfListTail])
										||N<-lists:seq(1, Concurency)];
						[] ->
							WorkFun=case proplists:get_value(output, RedOpts) of
										undefined -> 
											fun()->
													iter:foreach(
													  TotalIter,
													  fun(_)-> ok end)
											end;
										OutPipe when 
										  is_record(OutPipe,pipe) ->
											fun()->
													Pump=pump:new(TotalIter, OutPipe),
													pump:init(Pump)
											end
									end,
							[pgt:pg_spawn_worker(Node,WorkFun)
									   ||_N<-lists:seq(1, Concurency)]
					end,
					case RedFun of
						nil ->
							ok;
						_ ->
							start_ets_holder(PidList)
					end
			end),
	case RedFun of
		nil ->
			QPipe;
		_ ->
			%% Create pipe that forms a red fun args by Key i.e. {Key,[Iter]}
			pipe_mapper:new(
			  fun(Key) ->
					  {Key,[iter_mapper:new(
							  iter_ets_lookup:new(get_tid(), {MapId,Key}),
							  fun({K,V}) when K=={MapId,Key} ->
									  V
							  end)||MapId<-MapIdList]}
			  end, QPipe)
	end.

start_ets_holder(PidList) ->
	HPid=spawn_link(
	  fun()->
			  RefPidList=[{erlang:monitor(process, Pid),Pid}||Pid<-PidList],
			  [receive 
				   {'DOWN', Ref, process, Pid, _} ->
					   ok					
			   end||{Ref,Pid}<-RefPidList]
	  end),
	ets:give_away(get_tid(), HPid, give_to_mappers),
	HPid.
	
%% red_produce_init(ConsumerPid,RedFun,{Key,IterList},Quote) ->
%% 	IterState0=iter:open(RedFun(Key,IterList)),
%% 	red_produce_loop(ConsumerPid,IterState0,Quote)
%% 	ok.
%% red_produce_loop(ConsumerPid,IterState,Quote) ->
%% 	
%% 	ConsumerPid ! .

%% RedPid ! {mapper_finish,Id,self()},
%% 						receive
%% 							{acknowledged,RedPid} ->
%% 								ok
%% 						end
%% red_proc_rcv_loop(PipeState) ->
%% 	FinishMsg=?FINISH_MSG,
%% 	receive
%% 		{item,Id,MappedItem} ->
%% 			PipeState1=pipe:push(PipeState,{Id,MappedItem}),
%% 			red_proc_rcv_loop(PipeState1);
%% 		{mapper_finish,_Id,MapperPid} ->
%% 			MapperPid ! {acknowledged,self()},
%% 			red_proc_rcv_loop(PipeState);
%% 		FinishMsg ->
%% 			pipe:close(PipeState)
%% 	end.
red_proc_rcv_pump(Pipe) ->
	Pump=pump:new(red_proc_rcv_iter(), Pipe),
	pump:init(Pump).

red_proc_rcv_iter() ->
	iter_unzip:new(
	  iter:simple_immutable(
		fun()-> ?FINISH_MSG end,
		fun red_proc_rcv_item/1, 
		fun(_)-> ok end)).
red_proc_rcv_item(FinishMsg)->
	receive
		{item,DataItem} ->
			DataItem;
		{mapper_finish,_Id,MapperPid} ->
			MapperPid ! {acknowledged,self()},
			red_proc_rcv_item(FinishMsg);
		FinishMsg ->
			?EOF
	end.

new_accum_pipe(RedOpts) ->
	Ets=create_accum_ets(RedOpts),
	InserterPipe=pipe_ets_insert:new(Ets, 1000, pipe:terminator()),
	pipe_switch:new(
	  fun({Id,_MappedItem}) ->
			  Id
	  end, 
	  fun(Id)->
			  pipe_mapper:new(
				fun({Id1,{Key, Item}=_MappedItem}) when Id==Id1 ->
						{{Id1,Key},Item}
				end, InserterPipe)
	  end).

new_accum_key_iter() ->
	iter_unique:new(
	  iter_mapper:new(
		iter_ets_keys:new(get_tid()),
		fun({_Id1,Key}) ->
				Key
		end)).

create_accum_ets(RedOpts) ->
	EtsOpts=[protected|
			 case proplists:get_bool(unique,RedOpts) of
				 true ->
					 [bag];
				 _ ->
					 [duplicate_bag]
			 end],
	put('$accum_ets',Ets=ets:new('$accum_ets', EtsOpts)),
	Ets.

get_tid() ->
	case get('$accum_ets') of
		undefined ->
			throw(no_accum_ets_err);
		Tid ->
			Tid
	end.
					
	
%%====================================================================
%% eunit tests
%%====================================================================

%% RedSpec = {red, RedFun, [MapSpec]}
%% RedFun(Key,[Iter]) -> Iter1
%% MapSpec = {map, MapFun, Source}
%% MapFun(Item) -> MappedItem={Key,Item1} | [MappedItem] | MappedItemIter 
%% Source = Iter| {Node,Iter} | RedSpec
print_test() ->
	Self=self(),
	Spec={red, 
		  fun(Key,[Iter])->
				  {Is,ok}=iter:foreach(
							Iter,
							fun(Item)->
									io:format("~p ~p ~p~n",[self(),Item,Key]),
									Self ! asserted
							end),
				  iter:close(Is)
		  end, 
		  [{map,
			fun(Item) ->
					{erlang:phash2(Item),Item}
			end,
			iter:from_list([alfa,beta,gamma,delta,dzeta,eta,kappa,lambda,mu,nu,omicron,fi,hi,psi,omega])
		   }
		  ]},
	exec(Spec),
	?assert(receive_asserted(0)>0).
count_test() ->
	Self=self(),
	TestData=[{one,2},{two,3},{three,1},{four,1},{five,1},{six,1}],
	Spec={red, 
		  fun(Key,[Iter])->
				  {Is,Count}=iter:fold(
							   Iter,
							   fun(_Item,AccIn)->
									   AccIn+1
							   end, 0),
				  ?assertEqual(lists:keysearch(Key, 1, TestData),
							   {value,{Key,Count}}),
				Self ! asserted,
				  iter:close(Is)
		  end, 
		  [{map,
			fun(Item) ->
					{Item,Item}
			end,
			iter:from_list([Item1||{Item,Count}<-TestData,Item1<-lists:duplicate(Count, Item)])
		   }
		  ]},
	exec(Spec),
	?assert(receive_asserted(0)>0).
join_test() ->
	TestData1=[{1,"Vasya"},{2,"Petya"},{3,"Katya"}],
	TestData2=[{1,"Vasya1"},{2,"Petya1"},{3,"Katya1"},{1,"Vasya11"}],
	Self=self(),
	Spec={red, 
		  fun(Key,[Iter1,Iter2])->
				  Iter=iter_join_cprod:new(
						 Iter1, Iter2), 
				  Val=lists:sort(iter:to_list(Iter)),
				  Expec=lists:sort([[{Id1,Val1},{Id2,Val2}]||
									{Id1,Val1}<-TestData1,
									{Id2,Val2}<-TestData2,
									Id1==Id2,Id1==Key]),
				  ?assertEqual(Expec,Val),
				  Self ! asserted
		  end, 
		  [{map,
			fun({Id,_Name}=Item) ->
					{Id,Item}
			end,
			iter:from_list(TestData1)
		   },
		   {map,
			fun({Id,_Name}=Item) ->
					{Id,Item}
			end,
			iter:from_list(TestData2)
		   }],
		 [{concurrent,4}]},
	exec(Spec),
	?assert(receive_asserted(0)>0).
join3_test() ->
	Spec=
		{red,
		 fun(_Key,[Iter])->
				 iter:print(Iter)
		 end,
		 [{map,
		   fun([N1,N2])->
				   {N2,{N2,N1}}
		   end,
		   {red, 
			fun(_Key,[Iter1,Iter2])->
					iter_join_cprod:new(Iter1, Iter2)
			end, 
			[{map,
			  fun({Id,_Name}=Item) ->
					  {Id,Item}
			  end,
			  iter:from_list([{1,"Vasya"},{2,"Petya"},{3,"Katya"}])
			 },
			 {map,
			  fun({Id,_Name}=Item) ->
					  {Id,Item}
			  end,
			  iter:from_list([{1,"Vasya1"},{2,"Petya1"},{3,"Katya1"},{1,"Vasya11"}])
			 }]
		   ,[{concurrent,10}]}
		  }
		 ]
		},
	exec(Spec).
%% R J(Friend*FName)
%%  <-M
%%  <-M-R J(Name*Friend)
%%   	<-M
%%   	<-M
friends_join_test() ->
	NameList=[{1,"Vasya"},{2,"Petya"},{3,"Katya"},{4,"Sveta"},{5,"Olya"}],
	FriendList=[{1,3},{2,5},{3,4}],
	MapNames={map,
			  fun({Id,_Name}=Item) ->
					  {Id,Item}
			  end,
			  iter:from_list(NameList)
			 },
	MapFriends={map,
			  fun({Id,_IdF}=Item) ->
					  {Id,Item}
			  end,
			  iter:from_list(FriendList)
			 },
	JoinNameFriendId={red,
		 fun(_Key,[IterNames,IterFriendIds])->
				 iter_join_cprod:new(IterNames, IterFriendIds)
		 end,[MapNames,MapFriends]},
	MapNmFrId_ByFrId={map,
			  fun([{Id,_Name},{Id,FrId}]=Item) ->
					  {FrId,Item}
			  end,
			  JoinNameFriendId
			 },
	JoinNmFrNm={red,
		 fun(_Key,[Iter1,Iter2])->
				 iter_join_cprod:new(Iter1, Iter2)
		 end,[MapNmFrId_ByFrId,MapNames]},
	Spec=JoinNmFrNm,
	ResultList=iter:to_list(iter(Spec)),
	?assertEqual(3,length(ResultList)),
	?assertEqual(lists:sort([[[{Id,Name},{Id,FrId}],{FrId,FrName}]||
							 {Id,Name}<-NameList,{Id1,FrId1}<-FriendList,{FrId,FrName}<-NameList,
							 (Id==Id1 andalso 
								  FrId1==FrId)]),
				 lists:sort(ResultList)).

receive_asserted(N) ->
	receive
		asserted ->
			receive_asserted(N+1)
	after 0 ->
			N
	end.

for(Begin,End,Incr,_FoldFun,Acc0) when Begin<End andalso Incr<0 ->
	Acc0;
for(Begin,End,Incr,_FoldFun,Acc0) when Begin>End andalso Incr>0 ->
	Acc0;
for(Begin,End,Incr,FoldFun,Acc0) when 
  Incr/=0,is_integer(Incr),is_integer(Begin),
  is_integer(End),is_function(FoldFun, 2) ->
	for(Begin+Incr,End,Incr,FoldFun,FoldFun(Begin,Acc0)).

write_random_data(N,File) ->
	ok=file:delete(File),
	pipe:close(for(1,N,1,
				   fun(Idx,PipeState)->
						   pipe:push(PipeState,{Idx,{"some valuable data",random:uniform(N)}})
				   end,
				   pipe:open(pipe_disk_logger:new([{name,test},{file,File}])))).
write_random_data_zip(N,File) ->
	ok=file:delete(File),
	pipe:close(for(1,N,1,
				   fun(Idx,PipeState)->
						   pipe:push(PipeState,{Idx,{"some valuable data",random:uniform(N)}})
				   end,
				   pipe:open(pipe_zip:new(1000,pipe_disk_logger:new([{name,test},{file,File}]))))).

test_bulk_join_zip() ->
	File1="./test-mapred1.log",
	File2="./test-mapred2.log",
	write_random_data_zip(10000,File1),
	write_random_data_zip(15000,File2),
	Spec={red, 
		  fun(_Key,[Iter1,Iter2])->
				  iter_join_cprod:new(
					Iter1, Iter2)
		  end, 
		  [{map,
			fun({Id,_Data}=Item) ->
					{Id,Item}
			end,
			iter_unzip:new(iter_disk_log:new([{name,test1},{file,File1}])),
			[{zip,100}]
		   },
		   {map,
			fun({Id,_Data}=Item) ->
					{Id,Item}
			end,
			iter_unzip:new(iter_disk_log:new([{name,test2},{file,File2}])),
			[{zip,100}]
		   }],
		 [{concurrent,4}]},
	iter:print(iter(Spec)).
test_bulk_join() ->
	File1="./test-mapred1.log",
	File2="./test-mapred2.log",
	write_random_data(10000,File1),
	write_random_data(15000,File2),
	Spec={red, 
		  fun(_Key,[Iter1,Iter2])->
				  iter_join_cprod:new(
						   Iter1, Iter2)
		  end, 
		  [{map,
			fun({Id,_Data}=Item) ->
					{Id,Item}
			end,
			iter_disk_log:new([{name,test1},{file,File1}])
		   },
		   {map,
			fun({Id,_Data}=Item) ->
					{Id,Item}
			end,
			iter_disk_log:new([{name,test2},{file,File2}])
		   }],
		 [{concurrent,4}]},
	iter:print(iter(Spec)).
test_bulk_join2() ->
	File1="./test-mapred1.log",
	File2="./test-mapred2.log",
	write_random_data(10000,File1),
	write_random_data(15000,File2),
	Spec={red, 
		  fun(_Key,[Iter1,Iter2])->
				  Iter=iter_join_cprod:new(
						   Iter1, Iter2),
				  iter:print(Iter)
		  end, 
		  [{map,
			fun({_,{_,Id}}=Item) ->
					{Id,Item}
			end,
			iter_disk_log:new([{name,test1},{file,File1}])
		   },
		   {map,
			fun({_,{_,Id}}=Item) ->
					{Id,Item}
			end,
			iter_disk_log:new([{name,test2},{file,File2}])
		   }],
		 [{concurrent,4}]},
	exec(Spec).
iter_test() ->
	File1="./test-mapred1.log",
	File2="./test-mapred2.log",
	write_random_data(10000,File1),
	write_random_data(15000,File2),
	Spec={red, 
		  fun(_Key,[Iter1,Iter2])->
				  iter_join_cprod:new(
					  Iter1, Iter2)
		  end, 
		  [{map,
			fun({_,{_,Id}}=Item) ->
					{Id,Item}
			end,
			iter_disk_log:new([{name,test1},{file,File1}])
		   },
		   {map,
			fun({_,{_,Id}}=Item) ->
					{Id,Item}
			end,
			iter_disk_log:new([{name,test2},{file,File2}]),
			[{zip,100}]
		   }],
		  [{concurrent,4}]},
	iter:to_list(iter(Spec)).
iter_page_test() ->
	File1="./test-mapred1.log",
	File2="./test-mapred2.log",
	write_random_data(10000,File1),
	write_random_data(15000,File2),
	Spec={red, 
		  fun(_Key,[Iter1,Iter2])->
				  iter_join_cprod:new(
					  Iter1, Iter2)
		  end, 
		  [{map,
			fun({_,{_,Id}}=Item) ->
					{Id,Item}
			end,
			iter_disk_log:new([{name,test1},{file,File1}])
		   },
		   {map,
			fun({_,{_,Id}}=Item) ->
					{Id,Item}
			end,
			iter_disk_log:new([{name,test2},{file,File2}])
		   }],
		  [{concurrent,4}]},
	IsF=for(1, 10, 1, 
			fun(N,IterState)->
					{Is,Item}=iter:next(IterState),
					io:format("~p - ~p~n",[N,Item]),
					Is
			end, iter:open(iter(Spec))),
	iter:close(IsF).

nil_test() ->
	TestData=[alfa,beta,gamma,delta,dzeta,eta,kappa,lambda,mu,nu,omicron,fi,hi,psi,omega],
	Spec={red, 
		  nil, 
		  [{map,
			fun(Item) ->
					{erlang:phash2(Item),Item}
			end,
			iter:from_list(TestData)
		   }]},
	Result=lists:sort(iter:to_rlist(iter(Spec))),
	?assertEqual(lists:sort(TestData),Result).
nil1_test() ->
	TestData=[alfa,beta,gamma,delta,dzeta,eta,kappa,lambda,mu,nu,omicron,fi,hi,psi,omega],
	Spec={red, 
		  nil, 
		  [{map,
			fun(Item) ->
					[{1,Item},
					{2,Item},
					{3,Item},
					{4,Item}]
			end,
			iter:from_list(TestData)
		   }]},
	Result=lists:sort(iter:to_rlist(iter(Spec))),
	?assertEqual(lists:sort(TestData++TestData++TestData++TestData),Result).
nil2_test() ->
	TestData=[alfa,beta,gamma,delta,dzeta,eta,kappa,lambda,mu,nu,omicron,fi,hi,psi,omega],
	Spec={map,
		  fun(Item) ->
				  iter:from_list([{1,Item},
									   {2,Item},
									   {3,Item},
									   {4,Item}])
		  end,
		  iter:from_list(TestData)
		 },
	Result=lists:sort(iter:to_rlist(iter(Spec))),
	?assertEqual(lists:sort(TestData++TestData++TestData++TestData),Result).
after_nil_test() ->
	TestData=[alfa,beta,gamma,delta,dzeta,eta,kappa,lambda,mu,nu,omicron,fi,hi,psi,omega],
	Spec={map,
		  fun(Item)->
				  {erlang:phash2(Item),Item}
		  end,
		  {red, 
		  nil, 
		  [{map,
			fun(Item) ->
					iter:from_list([{1,Item},
					{2,Item},
					{3,Item},
					{4,Item}])
			end,
			iter:from_list(TestData)
		   }]}
		 },
	Result=lists:sort(iter:to_rlist(iter(Spec))),
	?assertEqual(lists:sort(TestData++TestData++TestData++TestData),Result).


